package main.scala.com.hu.api

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.hu.entity.SensorReading
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala._

/**
 * @Author: 22834
 * @Date: 2020/11/15 9:43
 * @Describe:
 */
object MyMysqlSinkDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputPath = "D:\\hjj\\I\\ideaWorkSpace\\hu\\flink-scala-learn\\src\\main\\resources\\data\\sersor.txt"

    val dataStream = env.readTextFile(inputPath).map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )

    //自定义mysql sink
    dataStream.addSink(new MyMysqlSinkFunc())

    env.execute("define mysql sink Demo")
  }
}

class MyMysqlSinkFunc() extends RichSinkFunction[SensorReading] {
  var conn: Connection = _
  var insertStat: PreparedStatement = _
  var updateStat: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "root")
    insertStat = conn.prepareStatement("insert into sensor (id,temperature) value(?,?)")
    updateStat = conn.prepareStatement("update sensor set temperature = ? where id=?")
  }

  override def invoke(value: SensorReading): Unit = {
    updateStat.setDouble(1, value.temperature)
    updateStat.setString(2, value.id)
    updateStat.execute()

    if (updateStat.getUpdateCount == 0) {
      insertStat.setString(1, value.id)
      insertStat.setDouble(2, value.temperature)
      insertStat.execute()
    }
  }

  override def close(): Unit = {
    updateStat.close()
    insertStat.close()
    conn.close()
  }
}
